package com.lrj.datastream.transform.agg;

import com.lrj.datastream.domain.ClickEvent;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author lrj
 * @date 2022/3/28 16:47
 */
public class TestReduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<ClickEvent> stream = env.fromElements(
                new ClickEvent("Mary", "./home", 1000L),
                new ClickEvent("Blob", "./cart", 2100L),
                new ClickEvent("Mary", "./product?id=100", 3000L),
                new ClickEvent("Mary", "./product?id=1", 3500L),
                new ClickEvent("Mary", "./product?id=23", 4200L),
                new ClickEvent("Blob", "./product?id=100", 4500L));
        // 泛型擦除导致 org.apache.flink.api.common.functions.InvalidTypesException.
        // In many cases lambda methods don't provide enough information
        stream.map((ClickEvent v)-> Tuple2.of(v,1L))
                .returns(new TypeHint<Tuple2<ClickEvent, Long>>(){})
                .keyBy(v->v.f0.getUser())
                .reduce((Tuple2<ClickEvent,Long> v1, Tuple2<ClickEvent,Long>v2) -> Tuple2.of(v1.f0,v1.f1+v2.f1))
                .print();
        env.execute(TestReduce.class.getName());
    }
}
